package com.rocketmq.producer.contorller;

import apache.rocketmq.v2.Message;
import jakarta.annotation.PostConstruct;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RestController;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Random;

/**
 * @program: awesome-spring
 * @description: 发送控制器
 * @author: Alex Wu
 * @createDate: 2025-02-11 09:46
 **/
@RestController
public class ProducerController {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @GetMapping(value = "/syncSend")
    public void syncSend() {
        int i = new Random().nextInt(100);
        SendResult sendResult = rocketMQTemplate.syncSend("niHao", "hello rocketmq " + i);
        System.out.println("Message sent: " + sendResult);
    }

    @GetMapping(value = "/syncSendDLQ")
    public void syncSendDLQ() {
        SendResult sendResult = rocketMQTemplate.syncSend("niHao", "dlq2");
        System.out.println("Message sent: " + sendResult);
    }

    @GetMapping(value = "/asyncSend")
    public void asyncSend() {
        int i = new Random().nextInt(100);
        String topic = "niHao";
        String message = "hello rocketmq " + i;
        rocketMQTemplate.asyncSend(topic, message, new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("Async message sent: " + message);
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("Async message error: " + e);
            }
        });
        System.out.println("Message sent: " + message);
    }


    @GetMapping(value = "/delayedSend")
    public void delayedSend() {
        int i = new Random().nextInt(100);
        String topic = "niHao";
        String messageContent = "hello rocketmq " + i;


        // 旧版仅支持延时等级
        // messageDelayLevel=1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h
        /*rocketMQTemplate.syncSend(topic, MessageBuilder.withPayload(messageContent).build(), 3000, 3);
        System.out.println("Delayed message sent: " + messageContent + " with delayLevel: " + 3);*/

        // 新版支持定时时间
        long delayMilliseconds = 1000 * 5;

        long deliverTime = System.currentTimeMillis() + (5 * 1000L); // 60 秒后

        // 定时
        rocketMQTemplate.syncSendDeliverTimeMills(topic, messageContent, deliverTime);

        long delayTime = 3000;
        // 延时
        rocketMQTemplate.syncSendDelayTimeMills(topic, messageContent, delayTime);

        System.out.println("Delayed message sent: " + messageContent + " with delayTime: " + delayMilliseconds);

        rocketMQTemplate.sendMessageInTransaction("niHao", MessageBuilder.withPayload("hello rocketmq " + i).build(), null);
    }


    @GetMapping(value = "/syncSendTransaction")
    public void syncSendTransaction() {
        String msg = "tx" + LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss"));
        System.out.println("start sendTransMessage : " + msg);

        TransactionSendResult result=rocketMQTemplate.sendMessageInTransaction("test-tx-rocketmq",
                MessageBuilder.withPayload(msg).build(), msg);

        //发送状态
        String sendStatus = result.getSendStatus().name();
        System.out.println("send tx message ! ");

        // 本地事务执行状态
        String localTxState = result.getLocalTransactionState().name();
        System.out.println("send tx message sendStatus: " + sendStatus + ",localTXState: " + localTxState);

    }


}
